package com.apps.sdses.flink141.udf;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.annotation.DataTypeHint;
import org.apache.flink.table.annotation.FunctionHint;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.functions.ScalarFunction;
import org.apache.flink.table.functions.TableFunction;
import org.apache.flink.types.Row;

/**
 * author: codejiwei
 * date: 2023/11/24
 * desc:
 **/
public class ScalarFunctionDemo {
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);


        tableEnv.createTemporaryFunction("HashFunction", HashFunction.class);

        //Source
        tableEnv.executeSql("CREATE TABLE sourceTable (\n" +
                "    id        BIGINT NOT NULL,\n" +
                "    geom_data        STRING,\n" +
                "    event_time   TIMESTAMP(3),\n" +
                "    temperature   INT NOT NULL, \n" +
                "    pressure   BIGINT\n" +
                ") WITH (\n" +
                "  'connector' = 'datagen', \n" +
                "  'number-of-rows' = '5', \n" +
//                "  'rows-per-second' = '1', \n" +
                "  'fields.id.min' = '1', \n" +
                "  'fields.id.max' = '5', \n" +
                "  'fields.geom_data.length' = '5', \n" +
                "  'fields.temperature.min' = '10', \n" +
                "  'fields.temperature.max' = '50', \n" +
                "  'fields.pressure.min' = '10', \n" +
                "  'fields.pressure.max' = '100'\n" +
                ")");

//        tableEnv.executeSql("CREATE TABLE sinkTable WITH ('connector' = 'print')\n" +
//                "LIKE sourceTable (EXCLUDING ALL)");
        tableEnv.executeSql("CREATE TABLE sinkTable(id BIGINT, temp_diff BIGINT) WITH ('connector' = 'print')");
        //Insert
        tableEnv.executeSql("insert into sinkTable select id, HashFunction(temperature, '100,200,300') from sourceTable where id in (1, 2)");



    }

//    @FunctionHint(input = @DataTypeHint("INT, STRING"))
    public static class HashFunction extends ScalarFunction {
        // take any data type and return INT
        public int eval(int o, String fixedValues) {
            String[] valuesArray = fixedValues.split(",");

            int sum = o;
            for (String value : valuesArray) {
                sum += Integer.parseInt(value.trim());
            }

            return sum;
        }
    }

    @FunctionHint(output = @DataTypeHint("ROW<word STRING, length INT>"))
    public static class SplitFunction extends TableFunction<Row> {
        public void eval(String str) {
            for (String s : str.split(" ")) {
                // use collect(...) to emit a row
                collect(Row.of(s, s.length()));
            }
        }
    }

}
